package com.roc.rocketmqconsumer.messageType;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 批量消息-生产者
 *
 * @author Roker
 * @date 2025/07/06
 */
public class BatchMsgConsumer {
    public static void main(String[] args) throws Exception {
        // 1.创建消息生产者，，指定消费者组（需和生产者逻辑匹配，保证订阅关系）
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("roc-consumer-group");
        // 2. 设置 NameServer 地址，要和生产者使用的一致
        consumer.setNamesrvAddr("127.0.0.1:9876");
        // 3. 订阅 Topic（需和生产者发送的 Topic 一致，这里示例是 "TOPIC_BATCH"）
        //    * 表示订阅该 Topic 下所有 Tag 的消息，也可指定具体 Tag 如 "TAG1 || TAG2"
        consumer.subscribe("TOPIC_BATCH", "*");
        // 4. 注册消息监听器，处理接收到的批量消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                // 遍历接收到的批量消息
                for (MessageExt msg : msgs) {
                    try {
                        // 解析消息体（按实际编码处理，这里假设是 UTF-8）
                        String msgBody = new String(msg.getBody(), "UTF-8");
                        System.out.println("消费消息：Topic=" + msg.getTopic()
                                + ", Tag=" + msg.getTags()
                                + ", 消息内容=" + msgBody);
                    } catch (Exception e) {
                        e.printStackTrace();
                        // 消息处理异常，返回 RECONSUME_LATER 告知 Broker 稍后重试
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                // 消息全部处理成功，返回 CONSUME_SUCCESS
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 5. 启动消费者
        consumer.start();
        System.err.println("消费者已启动，等待接收批量消息...");

        // （可选）让主线程不退出，保持监听状态。实际应用中可按需处理线程生命周期
        Thread.sleep(Long.MAX_VALUE);
    }
}
